package com.atguigu.day05;

import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.time.Duration;

public class Example10 {
    public static void main(String[] args) {
        var env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        var result = env
                // "a 1"
                // "a 2"
                // "a 10"
                .socketTextStream("localhost", 9999)
                // "a 1" ---> ("a", 1000L)
                .map(new MapFunction<String, Tuple2<String, Long>>() {
                    @Override
                    public Tuple2<String, Long> map(String in) throws Exception {
                        var arr = in.split(" ");
                        return Tuple2.of(
                                arr[0],
                                // 转换成毫秒时间戳
                                Long.parseLong(arr[1]) * 1000L
                        );
                    }
                })
                // 在map输出的数据流中插入水位线事件
                // `.assignTimestampsAndWatermarks`默认每隔200毫秒向下游发送一条水位线数据
                .assignTimestampsAndWatermarks(
                        // `Duration.ofSeconds(5)`表示设置 `最大延迟时间` 为5秒钟
                        WatermarkStrategy.<Tuple2<String, Long>>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                                .withTimestampAssigner(new SerializableTimestampAssigner<Tuple2<String, Long>>() {
                                    @Override
                                    public long extractTimestamp(Tuple2<String, Long> element, long recordTimestamp) {
                                        return element.f1; // 告诉flink事件时间戳是f1字段
                                    }
                                })
                )
                .process(new ProcessFunction<Tuple2<String, Long>, String>() {
                    @Override
                    public void processElement(Tuple2<String, Long> in, Context context, Collector<String> collector) throws Exception {
                        if (context.timestamp() < context.timerService().currentWatermark()) {
                            context.output(
                                    // 侧输出流的名字
                                    new OutputTag<String>("late-event"){},
                                    // 向侧输出流发送的数据
                                    "迟到数据：" + in
                            );
                        } else {
                            collector.collect("正常数据：" + in);
                        }
                    }
                });

        result.print("main");

        result.getSideOutput(new OutputTag<String>("late-event"){}).print("side");

        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
